package com.patsnap.data.npd.dw.etl.job.ods.sink;


import com.patsnap.data.npd.dw.etl.job.ods.dispatcher.DataLayer;
import com.patsnap.data.npd.dw.etl.job.ods.dispatcher.SinkRouterDispatcher;
import com.patsnap.data.npd.dw.etl.job.ods.serialization.PreOdsInfo;
import com.patsnap.data.npd.dw.etl.utils.FlinkDBOperatorUtils;
import com.patsnap.data.npd.dw.repository.DBOperator;
import com.patsnap.one.etl.flink.sink.BaseMybatisRichSinkFunction;
import org.apache.flink.configuration.Configuration;

import java.util.HashMap;
import java.util.Map;

public class OdsNpdRecordSink extends BaseMybatisRichSinkFunction<PreOdsInfo> {

    private final SinkRouterDispatcher<PreOdsInfo> sinkRouterDispatcher;

    private Map<String, Object> globalJobParameters;

    private DBOperator dbOperator;

    public OdsNpdRecordSink(DataLayer dataLayer) {
        sinkRouterDispatcher = new SinkRouterDispatcher<>(dataLayer);
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        globalJobParameters = new HashMap<>(getRuntimeContext().getExecutionConfig().getGlobalJobParameters().toMap());
        dbOperator = FlinkDBOperatorUtils.initDBOperator(sqlSessionFactoryMap);
    }

    @Override
    public void doInvoke(PreOdsInfo odsLegalCollectContext, Context context) throws Exception {
        sinkRouterDispatcher.dispatcher(odsLegalCollectContext, globalJobParameters, dbOperator);
    }
}
